package study.bigdata.flink.job;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;


// study.bigdata.flink.job.DemoJob
@Slf4j
public class DemoJob {

    @SneakyThrows
    public static void main(String[] args) {
        //
        StreamExecutionEnvironment exeEnv = StreamExecutionEnvironment.getExecutionEnvironment();

        //
        DataStreamSource<String> ds = exeEnv.socketTextStream("172.24.33.47", 9999);
        //
        AllWindowedStream<Tuple2<String, Integer>, TimeWindow> ws = ds.map(e -> {
                    log.info("这是一个测试日志");
                    return Tuple2.of(e, 1);
                })
                .returns(Types.TUPLE(Types.STRING, Types.INT))
                .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5)));
        //
        ws.sum(1).print();

        //
        exeEnv.execute();
    }
}
